package net.atomtree.util;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Admin;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.client.Table;
import org.apache.hadoop.hbase.util.Bytes;

public class HBaseQuery {

	/*
     * Scan�࣬�û��������scan()������ʵ����ֻ��Ҫ����HTable��getScaner()������
	 */


    /*
     * �������е��б�
     */
    public ArrayList<String> TableList(Connection conn) throws IOException {
        Admin hadmin = null;
        ArrayList<String> list = new ArrayList<String>();
        hadmin = conn.getAdmin();
        HTableDescriptor[] tablesdesc = hadmin.listTables();
        String tablename = null;
        for (HTableDescriptor t : tablesdesc) {
            tablename = t.getTableName().toString();
            list.add(tablename);
        }
        return list;
    }

    /*
     * ������Result���ת��Ϊjson�ַ���
     */
    public String resultToJSON(Result rs) {
        String rowkey = Bytes.toString(rs.getRow());
        String resultJson = "{" + "\"" + rowkey + "\"" + ":{";
        List<Cell> celllist = rs.listCells();
        for (Cell i : celllist) {
            String key = Bytes.toString(CellUtil.cloneQualifier(i));
            String value = Bytes.toString(CellUtil.cloneValue(i));
            String q = Bytes.toString(CellUtil.cloneFamily(i));
            String restr = "\"" + q + ":" + key + "\"" + ":" + "\"" + value
                    + "\",";
            resultJson += restr;
        }
        resultJson = resultJson.substring(0, resultJson.length() - 1) + "}}";

        return resultJson;
    }

    /*
     * ����hbase��׼�Ľ������result �������Ϊ Connection�� �� ���� �� ���� rowkey
     */
    public Result getRowResult(Connection conn, String tablename, String rowkey)
            throws IOException {
        Table table = null;
        table = conn.getTable(TableName.valueOf(tablename));
        Get get = new Get(Bytes.toBytes(rowkey));
        Result result = table.get(get);
        return result;
    }

    /*
     * 1. ����json�ַ��� ������� Connection�� ���� rowkey ��������
     *
     * conn HBase����ʵ�� tablename Ҫ��ѯ��hbase���� rowkey ��ѯ��rowkey
     *
     * ���صĽ������resultToJSON()��װ��json�ַ���
     */
    public String getRowJSON(Connection conn, String tablename, String rowkey)
            throws IOException {
        Table table = null;
        table = conn.getTable(TableName.valueOf(tablename));
        Get get = new Get(Bytes.toBytes(rowkey));
        Result result = table.get(get);
        String resultJson = resultToJSON(result);
        // System.out.println(resultJson);

        return resultJson;
    }

	/*
	 * Scan������ȡ���� 1. ����Ӳ�������Scan 2. ��Ӳ�����S�в�������
	 */

    public ArrayList<String> ScanTable(Connection conn, String tablename)
            throws IOException {
        Table table = null;
        ArrayList<String> list = new ArrayList<String>();
        table = conn.getTable(TableName.valueOf(tablename));
        Scan scan = new Scan();
        ResultScanner rs = table.getScanner(scan);
        String restr;
        if (rs != null) {
            for (Result result : rs) {
                restr = resultToJSON(result);
                list.add(restr);
            }
        } else {
            list.add("NULL");
        }

        return list;
    }

    /*
     * Scan������ȡ����   �����ʼrowkey
     */
    public ArrayList<String> ScanTable(Connection conn, String tablename, String startRowkey)
            throws IOException {
        Table table = null;
        ArrayList<String> list = new ArrayList<String>();
        table = conn.getTable(TableName.valueOf(tablename));
        Scan scan = new Scan();
        scan.setStartRow(Bytes.toBytes(startRowkey));
        ResultScanner rs = table.getScanner(scan);
        String restr;
        if (rs != null) {
            for (Result result : rs) {
                restr = resultToJSON(result);
                list.add(restr);
            }
        } else {
            list.add("NULL");
        }

        return list;
    }

    public ArrayList<String> ScanTable(Connection conn, String tablename, String startRowkey, String stopRowkey)
            throws IOException {
        Table table = null;
        ArrayList<String> list = new ArrayList<String>();
        table = conn.getTable(TableName.valueOf(tablename));
        Scan scan = new Scan();
        scan.setStartRow(Bytes.toBytes(startRowkey));
        scan.setStopRow(Bytes.toBytes(stopRowkey));
        ResultScanner rs = table.getScanner(scan);
        String restr;
        if (rs != null) {
            for (Result result : rs) {
                restr = resultToJSON(result);
                list.add(restr);
            }
        } else {
            list.add("NULL");
        }

        return list;
    }
	
	
	
	
	
	

	/*
	 * ���Ʊ��show���ܣ�չ��ĳ�ű��ǰn�м�¼ ��� conn HBase����ʵ�� tablename Ҫ��ѯ��hbase���� num
	 * int���ͣ�Ҫ���ص�����
	 * 
	 * ���صĽ������resultToJSON()��װ��json�ַ���
	 */

    public ArrayList<String> showTable(Connection conn, String tablename,
                                       int num) throws IOException {
        Table table = null;
        ArrayList<String> list = new ArrayList<String>();
        table = conn.getTable(TableName.valueOf(tablename));
        Scan scan = new Scan();
        ResultScanner rscan = table.getScanner(scan);
        int i = 0;
        String oneresstr = null;
        for (Result rs : rscan) {
            if (i > num) {
                break;
            }
            oneresstr = resultToJSON(rs);
            list.add(oneresstr);
            i++;
        }
        return list;
    }

	/*
	 * ����������ر������еļ�¼��
	 */

    public static void main(String[] args) throws IOException {
        CreateConnection hc = new CreateConnection();
        Connection conn = hc.getConn();

        HBaseQuery hq = new HBaseQuery();
        ArrayList<String> res = hq.ScanTable(conn, "hbase_test", "rk213", "rk213");
        for (String str : res) {
            System.out.println(str);
        }


        // HBaseTableManage ht = new HBaseTableManage() ;
        // ArrayList<String> tablelist = ht.TableList(conn);
        // for(String d : tablelist){
        // System.out.println(d);
        // }

    }

}
